package com.nami.client.qos;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.nami.basic.enums.protocol.UpCMDEnum;
import com.nami.basic.packet.Packet;
import com.nami.client.cache.ClientChannelCache;
import com.nami.client.utils.JsonUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;


@Slf4j
public class QosReceiver {

    private static final QosReceiver INSTANCE = new QosReceiver();
    private ScheduledExecutorService scheduledExecutorService;
    private ExecutorService taskExecutorService;

    private static Multimap<String, QosItem> map = Multimaps.synchronizedMultimap(ArrayListMultimap.create());

    private final AtomicBoolean isRunning = new AtomicBoolean(false);

    public static QosReceiver getInstance() {
        return INSTANCE;
    }

    private QosReceiver() {
        scheduledExecutorService = new ScheduledThreadPoolExecutor(1
            , ThreadFactoryBuilder.create()
            .setNamePrefix("client-qos-handler-")
            .setDaemon(true)
            .setUncaughtExceptionHandler(((thread, throwable) ->
                log.error("[client-qos-handler-task] ThreadPool {} got exception", thread, throwable)))
            .build());
        scheduledExecutorService.scheduleAtFixedRate(this::doBatchSend, 1, 1, TimeUnit.SECONDS);


        taskExecutorService = new ThreadPoolExecutor(
            4, // 核心线程数
            8, // 最大线程数
            60, // 空闲线程存活时间
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(20), // 任务队列
            ThreadFactoryBuilder.create()
                .setNamePrefix("client-qos-task-")
                .setDaemon(true)
                .setUncaughtExceptionHandler(((thread, throwable) ->
                    log.error("[client-qos-task] ThreadPool {} got exception", thread, throwable)))
                .build(),
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
    }

    public void doBatchSend() {
        if (isRunning.getAndSet(true)) {
            return;
        }
        try {
            log.info("----------------------------{}----------------------------------------------", System.identityHashCode(this));
            synchronized (map) {
                Iterator<Map.Entry<String, Collection<QosItem>>> iterator = map.asMap().entrySet().iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, Collection<QosItem>> next = iterator.next();
                    String channelKey = next.getKey();
                    List<QosItem> qosItems = (List<QosItem>) next.getValue();

                    if (CollUtil.isNotEmpty(qosItems)) {
                        taskExecutorService.submit(() -> {
                            sendQosPacket(channelKey, qosItems);
                            log.info("[client-qos] 批量发送QoS, size={}", qosItems.size());
                        });
                    }
                    iterator.remove();

                }
            }

        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("[qos-client] 批量发送出问题了 msg={}", var3.getMessage());
        } finally {
            isRunning.set(false);
        }

    }

    private void sendQosPacket(String channelKey, List<QosItem> qosItems) {
        QosItem firstItem = qosItems.get(0);
        List<String> msgIds = qosItems.stream()
            .filter(item -> item != null && StringUtils.isNotBlank(item.getMsgId()))
            .map(QosItem::getMsgId)
            .collect(Collectors.toList());

        TimeInterval timer = DateUtil.timer();
        Packet qosPacket = createQosPacket(firstItem, msgIds);
        Channel channel = ClientChannelCache.getChannel(channelKey);
        if (channel == null) {
            return;
        }
        channel.writeAndFlush(qosPacket);

        if (timer.interval() > 100) {
            log.warn("[qos] 发送QoS耗时较长, use={}ms", timer.interval());
        }
    }

    private Packet createQosPacket(QosItem qosItem, List<String> msgIds) {
        Packet packet = new Packet();
        packet.setCmd(UpCMDEnum.QOS.getCode());
        packet.setVersion((byte) 1);
        packet.setEncrypt((byte) 1);
        packet.setSerial((byte) 10);

        UpMsgQosReq qosRequest = new UpMsgQosReq();
        qosRequest.setSenderId(qosItem.getSenderId());
        qosRequest.setReceiverId(qosItem.getReceiverId());
        qosRequest.setMsgIds(msgIds);
        qosRequest.setConverType("c2g");
        packet.setBody(JsonUtils.toJsonByte(qosRequest));

        return packet;
    }


    public static void add(String channelKey, QosItem qosItem) {
        map.put(channelKey, qosItem);
    }

    public void shutdown() {
        if (scheduledExecutorService != null) {
            scheduledExecutorService.shutdown();
            try {
                if (!scheduledExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduledExecutorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduledExecutorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        if (taskExecutorService != null) {
            taskExecutorService.shutdown();
            try {
                if (!taskExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    taskExecutorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                taskExecutorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

}
